package com.distributer.delayqueue.delayqueue.event;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.common.eventbus.AsyncEventBus;

public class EventBus {

	private com.google.common.eventbus.EventBus eventBus = null;
	
	private AtomicBoolean isRunning = new AtomicBoolean();
	
	private EventBus(){
		ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<Runnable>(1024);
		eventBus = new  AsyncEventBus(new ThreadPoolExecutor(4,
				8, 
				60,
				TimeUnit.SECONDS, 
				workQueue,
				new ThreadFactory() {
					private  AtomicInteger number = new AtomicInteger();
					private  String threadName = "job-listener-thread-%s";
					@Override
					public Thread newThread(Runnable r) {
						Thread thread = new Thread(r);
						thread.setName(String.format(threadName, number.incrementAndGet()));
						thread.setDaemon(true);
						return thread;
					}
				}));
	}
	
	public void register(JobEventListener jobEventListener){
		if (isRunning.compareAndSet(false, true)) {
			eventBus.register(jobEventListener);
		}
	}
	
	public void unRegister(JobEventListener jobEventListener){
		if (isRunning.get()) {
			eventBus.unregister(jobEventListener);
		}
	}
	
	public void postEvent(JobEvent jobEvent){
		if (isRunning.get()) {
			eventBus.post(jobEvent);
		}
	}
	
	public static EventBus getInstance(){
		return EventBusInstance.getInstance();
	}
	
	private static class EventBusInstance{
		private static EventBus bus;
		static{
			bus = new EventBus();
		}
		private EventBusInstance(){}
		
		public static EventBus getInstance(){
			return bus;
		}
	}
}
